package com.rainman.simpleredis.core.simplemq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.rainman.simpleredis.core.config.SimpleRedisConfig;
import com.rainman.simpleredis.core.distributed_lock.SimpleDistributedLock;
import com.rainman.simpleredis.core.scheduled.SimpleScheduled;
import com.rainman.simpleredis.core.simplemq.sch_worker.SerialServerSchWorker;
import com.rainman.simpleredis.core.utils.DelayList;

import java.time.Duration;
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.TimeUnit;


/**
 * 队列消息
 * 基于redis的消息发布订阅机制与list数据类型
 *
 */
public class SimpleMqSerialServer {


    private final SimpleRedisConfig simpleRedisConfig;
    private final SimpleDistributedLock simpleDistributedLock;

    private final DelayList delayList;

    public final String SerialHead;//SR0.Ori.SerialMq:
    public final String SerialIndex;//SR0.Index.SerialMq:
    private final String SerialDHead;//SR0.Dead.SerialMq:

    //分布式锁头数据
    public  final String DealLockTypeHead="SimpleRedis.SerialDealType:";

    public SimpleMqSerialServer(SimpleRedisConfig simpleRedisConfig
            , SimpleDistributedLock simpleDistributedLock
            , SimpleScheduled simpleScheduled){

        this.simpleRedisConfig = simpleRedisConfig;
        this.simpleDistributedLock=simpleDistributedLock;
        delayList=new DelayList();

        String keyHead=simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()+".";//SR0.
        SerialIndex=keyHead+"Index.SerialMq:";//SR0.Index.SerialMq:
        SerialHead=keyHead+"Ori.SerialMq:";//SR0.Ori.SerialMq:
        SerialDHead=keyHead+"Dead.SerialMq:";//SR0.Dead.SerialMq:

        simpleScheduled.addWorker(new SerialServerSchWorker("SimpleMqSerial.ScanData", this::scanData));

    }

    /**
     * 主要保证存在的队列有在消费
     * 消息以索引消息为准
     * 如果原始消息不在索引中被索引到将不会被处理
     *
     */
    private void scanData() {

        Set<String> keys = simpleRedisConfig.getRedisInterface().keys(SerialIndex + "*");//SR0.Index.SerialMq:*

        //处理各个索引队列
        for(String key:keys){//单个key示例SR0.Index.SerialMq:类型值
            String[] split = key.split(":");
            String type=split[1];

            Boolean durDeal = simpleDistributedLock.hasLock(DealLockTypeHead + type);
            //是否在处理中
            if(durDeal){
                continue;
            }

            System.out.println("SerialServer 兜底生效 可忽略"+type);
            publishMQSerialTypeTig(type);

        }

    }

    /**
     * 发布订阅消息触发
     * 某类型队列有需要尝试读取消息进行处理
     * 用时只有一个线程可以执行读取
     *
     * @param key
     */
    public void readOne(String key){

        String type=key.split(":")[1];

        SimpleMqWorkerInterface curWorker=null;
        for(SimpleMqWorkerInterface worker: simpleRedisConfig.getMqWorkerInterfaces()){
            if(worker.matchTargetType(type)){
                curWorker=worker;
            }
        }
        //判断是否有消费者
        if(curWorker==null){
            return;
        }

        //抢之前看下自己状态,上次有异常状态的晚点抢夺处理权限
        boolean b = delayList.hasKey(type);
        if(b){
            try {
                Thread.sleep(20L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        Boolean dLockAdd = simpleDistributedLock.dLockAdd(DealLockTypeHead +type, 40000L);
        if(!dLockAdd){
            return;
        }
        if(b){
            delayList.remove(type);
        }

        //抢夺处理权限成功，以下是消息处理

        //取消息
        String targetId= simpleRedisConfig.getRedisInterface().index(key,0);

        if(targetId==null){//没有消息了
            simpleDistributedLock.dLockRemove(DealLockTypeHead +type);

            //检测有没有原始消息不在索引中，确认3次后 进死信
            Set<String> keys = simpleRedisConfig.getRedisInterface().keys(SerialHead+type + ":*");

            for(String notIndexKey:keys) {//处理各个未索引值

                String value = simpleRedisConfig.getRedisInterface().getValue(notIndexKey);
                JSONObject jsonObject = JSONUtil.parseObj(value);

                Integer noIndexCount = jsonObject.getInt("noIndexCount", 0);
                if(noIndexCount>=3){

                    String[] split = notIndexKey.split(":");

                    jsonObject.set("errMsg","未被索引管理消息");

                    simpleRedisConfig.getRedisInterface().delete(SerialHead+type+":"+split[2]);
                    publishMQSerialDead(type,split[2],jsonObject.toString());
                }else {
                    jsonObject.set("noIndexCount",noIndexCount+1);
                    simpleRedisConfig.getRedisInterface().set(notIndexKey,jsonObject.toString());
                }

            }

            return;
        }

        dealOne(type,targetId,curWorker);

    }

    /**
     * 从单个队列的左边取一条消息
     * 并处理
     *
     */
    public void dealOne(String type,String targetId,SimpleMqWorkerInterface curWorker){

        String value = simpleRedisConfig.getRedisInterface().getValue(SerialHead  + type + ":" + targetId);

        if(value==null){
            simpleDistributedLock.dLockRemove(DealLockTypeHead + type);
            simpleRedisConfig.getRedisInterface().leftPop(SerialIndex+type, 1, TimeUnit.SECONDS);
            publishMQSerialTypeTig(type);
            return;
        }


        JSONObject msgObject = JSONUtil.parseObj(value);

        if(simpleRedisConfig.getVer()<msgObject.getLong("ver")){
            //高版本的消息。不会被低版本的消费者消费, 放弃抢夺权限，等待下一个兜底触发
            delayList.add(type);
            simpleDistributedLock.dLockRemove(DealLockTypeHead + type);
            publishMQSerialTypeTig(type);
            return;
        }

        WorkerReturn msg = new WorkerReturn(WorkerReturn.StateRetryImm);

        try{

            msg=curWorker.doWork(type, targetId, msgObject.getStr("msg"));


        }finally {

            boolean tig=true;

            if (msg==null||WorkerReturn.StateSuc.equals(msg.getState())){
                //成功
                simpleRedisConfig.getRedisInterface().leftPop(SerialIndex+type, 1, TimeUnit.SECONDS);
                simpleRedisConfig.getRedisInterface().delete(SerialHead+type+":"+targetId);

            }else if (WorkerReturn.StateRetry.equals(msg.getState())){

                delayList.add(type);
                simpleDistributedLock.dLockRemove(DealLockTypeHead +type);
                tig=false;

            }else if (WorkerReturn.StateRetryImm.equals(msg.getState())){
                //立即重试

                delayList.add(type);

            }else {//进死信

                msgObject.set("errMsg",msg.getFailMsg());
                //新增死信记录

                publishMQSerialDead(type,targetId,msgObject.toString());

                simpleRedisConfig.getRedisInterface().leftPop(SerialIndex+type, 1, TimeUnit.SECONDS);
                simpleRedisConfig.getRedisInterface().delete(SerialHead+type+":"+targetId);

            }

            if(tig){

                simpleDistributedLock.dLockRemove(DealLockTypeHead +type);
                publishMQSerialTypeTig(type);
            }
        }


    }

    /**
     * 记录原始消息
     *
     * @param targetType
     * @param targetId
     * @param msgStr
     */
    public void publishMQSerialOri(String targetType,  String targetId,String msgStr  ) {

        if(targetType.contains(":")){
            throw new RuntimeException("targetType不能包含非法字符 : ");
        }

        JSONObject jsonObject=new JSONObject();
        jsonObject.set("time",System.currentTimeMillis());
        jsonObject.set("msg",msgStr);
        jsonObject.set("para",false);
        jsonObject.set("ver", simpleRedisConfig.getVer());

        //原始数据
        //SR0.Ori.SerialMq:1000:1
        simpleRedisConfig.getRedisInterface().set(SerialHead+targetType+":"+targetId,jsonObject.toString());

        publishMQSerialTig(targetType,targetId);

    }


    private void publishMQSerialTig(String targetType,String targetId ) {

            simpleRedisConfig.getRedisInterface().rightPush(
                    SerialIndex+targetType
                    ,targetId+""
            );
        publishMQSerialTypeTig(targetType);

    }

    private void publishMQSerialTypeTig(String targetType) {//触发一个可读取

        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()
                ,SerialIndex+targetType
        );

    }


    protected void publishMQSerialDead(String targetType,  String targetId,String valueStr) {//死信保留时间为15天

        simpleRedisConfig.getRedisInterface()
                .set(SerialDHead+targetType+":"+targetId,valueStr, Duration.ofMillis(1000L*3600*24*15));

    }

}
